package com.tech.common

import com.tech.config.ApplicationConfig
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties

object KafkaUtils {

  val props = new Properties

  props.put("bootstrap.servers", ApplicationConfig.KAFKA_BROKER_LIST) // broker 如果有多个,中间使用逗号分隔
  props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  val producer = new KafkaProducer[String, String](props)

  def write2Kafka(topic: String, msg: String): Unit = {
    val producerRecord = new ProducerRecord[String, String](topic, msg)
    producer.send(producerRecord).get()
  }

  def getOffsetFromKudu(topics: String): String = {
    // 初始化一个kafka consumer, 用来获取topic相关信息
    val prop = new Properties
    prop.put("bootstrap.servers", ApplicationConfig.KAFKA_BROKER_LIST)
    prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val kafkaConsumer = new KafkaConsumer[String, String](prop)
    // 获取所有 topic
    val topicList = kafkaConsumer.listTopics()
    // 判断 topic 是否存在
    val specifyTopics: Array[String] = topics.replace(" ", "").split(",")
    specifyTopics.foreach {
      topic =>
        if (!topicList.containsKey(topic))
          throw new Exception(s"The specified topic: $topic does not exist")
    }
    val map = KuduUtil.getOffset()
    // 拼接topic的partition offset信息
    var result = "{"

    specifyTopics.foreach(
      topic => {
        val iterator = kafkaConsumer.partitionsFor(topic).iterator()
        var i: String = "\"" + topic + "\":{"
        while (iterator.hasNext) {
          val partition = iterator.next().partition()
          val offset = if (map.contains(topic + "-" + partition)) {
            map(topic + "-" + partition)
          } else {
            -2
          }
          i += "\"" + partition + "\":" + offset + ","
        }
        i = i.substring(0, i.length - 1)
        i += "}"
        result += i
        result += ","
      }
    )
    result = result.substring(0, result.length - 1)
    result += "}"
    kafkaConsumer.close()
    result
  }
}
